package com.spark.core.transformation

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * join  会产生shuffle
 * （K,V）格式的RDD和（K,W）格式的RDD按照key相同join 得到（K,(V,W)）格式的数据。
 *
 * 如果nameRDD和scoreRDD的分区数相同，那么joinRDD的分区数也将与它们相同。
 * 这是因为Spark会尽量保持父RDD的分区结构，以提高性能。
 *
 * 如果nameRDD和scoreRDD的分区数不同，那么Spark会重新分区其中一个RDD，使其分区数与另一个RDD相匹配，
 * 然后执行join操作。这通常会导致数据重分布和数据移动，因此需要谨慎选择分区数，以避免不必要的性能开销。
 *
 * 总之，joinRDD的分区数与nameRDD和scoreRDD中分区数较多的那个RDD的分区数相关联，
 * 因为Spark通常会尽量避免数据的重分区。要更好地管理分区数和性能，
 * 可以在join操作之前使用repartition或coalesce等转换操作来调整RDD的分区数。
 *
 * 上面关于分区数的介绍对 leftOuterJoin, rightOuterJoin, fullOuterJoin 同样适用
 */
object Demo8_join {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("join")
    conf.setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val nameRDD = sc.parallelize(List[(String, String)](("zhangsan", "female"), ("zhangsan", "new"),
      ("lisi", "male"), ("wangwu", "female"), ("zhaoliu", "male")), 4)
    val scoreRDD = sc.parallelize(List[(String, Int)](("zhangsan", 18), ("zhangsan", 20),
      ("lisi", 19), ("wangwu", 20), ("tianqi", 25)), 3)

    val joinRdd: RDD[(String, (String, Int))] = nameRDD.join(scoreRDD)

    println(joinRdd.getNumPartitions)
    joinRdd.foreach(println)
  }
}
